package cn.monkey.transport.netty;

import io.netty.bootstrap.AbstractBootstrap;
import io.netty.channel.*;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;

import java.net.SocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Function;

/**
 * @param <BOOTSTRAP>
 * @param <CONF>
 * @param <B>
 * @param <C>
 */
public abstract class Transport<BOOTSTRAP extends AbstractBootstrap<BOOTSTRAP, C>, CONF extends TransportConfig<BOOTSTRAP, CONF, B, C>,
        B extends TransportConfig<BOOTSTRAP, CONF, B, C>.Builder, C extends Channel> {

    private static final Logger log = LoggerFactory.getLogger(Transport.class);
    protected final CONF config;
    protected final Map<SocketAddress, C> channelMap;
    protected final BOOTSTRAP bootstrap;

    public Transport(CONF config) {
        this.config = this.fixedConfig(config);
        this.channelMap = new ConcurrentHashMap<>();
        this.bootstrap = this.init(this.newBootStrap());
    }

    public Mono<Void> close() {
        ShutDown transportShutDown = this.config.shutDown();
        return transportShutDown.apply(this);
    }

    public Mono<Void> close(SocketAddress socketAddress) {
        Channel remove = this.channelMap.remove(socketAddress);
        if (remove == null) {
            log.warn("socket: {} channel is not exists", socketAddress);
            return Mono.error(new NullPointerException());
        }
        ChannelFuture channelFuture = remove.close();
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(future.cause());
        }));
    }

    protected abstract CONF fixedConfig(CONF config);

    protected abstract BOOTSTRAP newBootStrap();

    protected BOOTSTRAP init(BOOTSTRAP bootstrap) {
        bootstrap = this.bindEventLoopGroup(bootstrap);
        bootstrap = this.bindChannelFactory(bootstrap);
        bootstrap = this.bindChannelAttributes(bootstrap);
        bootstrap = this.bindChannelOptions(bootstrap);
        bootstrap = this.doOnChannelInit(bootstrap);
        return this.customizerBootStrap(bootstrap);
    }

    protected BOOTSTRAP customizerBootStrap(BOOTSTRAP bootstrap) {
        Function<BOOTSTRAP, BOOTSTRAP> bootstrapCustomizer = this.config.bootstrapCustomizer();
        if (bootstrapCustomizer != null) {
            return bootstrapCustomizer.apply(bootstrap);
        }
        return bootstrap;
    }

    protected abstract BOOTSTRAP bindEventLoopGroup(BOOTSTRAP bootstrap);

    protected BOOTSTRAP bindChannelFactory(BOOTSTRAP bootstrap) {
        return bootstrap.channelFactory(this.config.channelFactory());
    }

    protected BOOTSTRAP bindChannelAttributes(BOOTSTRAP bootstrap) {
        Map<AttributeKey<?>, ?> channelAttributes = this.config.channelAttributes();
        if (channelAttributes != null && !channelAttributes.isEmpty()) {
            for (Map.Entry<AttributeKey<?>, ?> e : channelAttributes.entrySet()) {
                @SuppressWarnings("unchecked") AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
                bootstrap.attr(key, e.getValue());
            }
        }
        return bootstrap;
    }

    protected static Future<?> shutDownEventLoopGroup(EventLoopGroup eventLoopGroup) {
        return eventLoopGroup.shutdownGracefully();
    }

    protected BOOTSTRAP bindChannelOptions(BOOTSTRAP bootstrap) {
        Map<ChannelOption<?>, ?> channelOptions = this.config.channelOptions();
        if (channelOptions != null && !channelOptions.isEmpty()) {
            for (Map.Entry<ChannelOption<?>, ?> e : channelOptions.entrySet()) {
                @SuppressWarnings("unchecked") ChannelOption<Object> key = (ChannelOption<Object>) e.getKey();
                bootstrap.option(key, e.getValue());
            }
        }
        return bootstrap;
    }

    protected abstract BOOTSTRAP doOnChannelInit(BOOTSTRAP bootstrap);
}
